package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.LearningPlayBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsPlayChapterVideoWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsPlayChapterVideoWindow().start(
                10089,
                4,
                "dws_play_chapter_video_window",
                Constant.TOPIC_DWD_TRAFFIC_APPVIDEO
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
//        kafkaStrDS.print();
        SingleOutputStreamOperator<LearningPlayBean> beanDS = kafkaStrDS.map(
                new MapFunction<String, LearningPlayBean>() {
                    @Override
                    public LearningPlayBean map(String value) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(value);
                        return LearningPlayBean.builder()
                                .videoId(jsonObj.getJSONObject("appVideo").getString("video_id"))
                                .userId(jsonObj.getJSONObject("common").getString("uid"))
                                .playSec(jsonObj.getJSONObject("appVideo").getLong("play_sec"))
                                .playCt(1L)
                                .ts(jsonObj.getLong("ts"))
                                .build();
                    }
                }
        );
//        beanDS.print();
        //根据userId分组
        KeyedStream<LearningPlayBean, String> uidKeyedDS = beanDS.keyBy(LearningPlayBean::getUserId);
        //标记独立用户
        SingleOutputStreamOperator<LearningPlayBean> uuBeanDS = uidKeyedDS.process(
                new KeyedProcessFunction<String, LearningPlayBean, LearningPlayBean>() {
                    //状态
                    private ValueState<String> lastvakueStateDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastvakueStateDate", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastvakueStateDate = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(LearningPlayBean learningPlayBean, KeyedProcessFunction<String, LearningPlayBean, LearningPlayBean>.Context context, Collector<LearningPlayBean> collector) throws Exception {
                        String value = lastvakueStateDate.value();
                        String date = DateFormatUtil.tsToDate(learningPlayBean.getTs());
                        if (value == null || value.compareTo(date) < 0) {
                            learningPlayBean.setPlayUuCt(1L);
                            lastvakueStateDate.update(date);
                        } else {
                            learningPlayBean.setPlayUuCt(0L);
                        }
                        collector.collect(learningPlayBean);
                    }
                }
        );
        //设置水位线
        SingleOutputStreamOperator<LearningPlayBean> withWaterMarkDS = uuBeanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<LearningPlayBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<LearningPlayBean>() {
            @Override
            public long extractTimestamp(LearningPlayBean learningPlayBean, long l) {
                return learningPlayBean.getTs();
            }
        }));
        //根据VideoId进行分组(各章节)
        KeyedStream<LearningPlayBean, String> videoIdKeyedDS = withWaterMarkDS.keyBy(LearningPlayBean::getVideoId);
        //开窗10s窗口
        WindowedStream<LearningPlayBean, String, TimeWindow> windowDS = videoIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //筛选每个章节的播放次数、累计播放时长和观看人数，并补充窗口开始时间结束时间和当前日期
        SingleOutputStreamOperator<LearningPlayBean> reduceDS = windowDS.reduce(
                new ReduceFunction<LearningPlayBean>() {
                    @Override
                    public LearningPlayBean reduce(LearningPlayBean value1, LearningPlayBean value2) throws Exception {
                        value1.setPlayCt(value1.getPlayCt() + value2.getPlayCt());
                        value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
                        value1.setPlayUuCt(value1.getPlayUuCt() + value2.getPlayUuCt());
                        return value1;
                    }
                },
                new ProcessWindowFunction<LearningPlayBean, LearningPlayBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<LearningPlayBean, LearningPlayBean, String, TimeWindow>.Context context, Iterable<LearningPlayBean> elements, Collector<LearningPlayBean> out) throws Exception {
                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());
                        LearningPlayBean bean = elements.iterator().next();
                        bean.setStt(stt);
                        bean.setEdt(edt);
                        bean.setCurDate(curDate);
                        out.collect(bean);
                    }
                }
        );
//        reduceDS.print("reduceDS->");
        //补充章节id
        SingleOutputStreamOperator<LearningPlayBean> chapterIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<LearningPlayBean>() {

                    @Override
                    public void addDims(LearningPlayBean bean, JSONObject dimJsonObj) {
                        bean.setChapterId(dimJsonObj.getString("chapter_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getRowKey(LearningPlayBean bean) {
                        return bean.getVideoId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //补充章节名称
        SingleOutputStreamOperator<LearningPlayBean> resultDS = AsyncDataStream.unorderedWait(
                chapterIdDS,
                new DimAsyncFunction<LearningPlayBean>() {
                    @Override
                    public void addDims(LearningPlayBean bean, JSONObject dimJsonObj) {
                        bean.setChapterName(dimJsonObj.getString("chapter_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_chapter_info";
                    }

                    @Override
                    public String getRowKey(LearningPlayBean bean) {
                        return bean.getChapterId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        resultDS.print();
        resultDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_play_chapter_video_window"));
    }
}
